package com.atguigu.gmall.realtime.common.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.bean.DimJoinFunction;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import com.atguigu.gmall.realtime.common.util.RedisUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

/**
 * @author Felix
 * @date 2024/10/14
 */
public abstract class DimMapFunction<T> extends RichMapFunction<T, T> implements DimJoinFunction<T> {

    private Connection hbaseConn;
    private Jedis jedis;

    @Override
    public void open(Configuration parameters) throws Exception {
        hbaseConn = HBaseUtil.getHBaseConnection();
        jedis = RedisUtil.getJedis();
    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeHBaseConnection(hbaseConn);
        RedisUtil.closeJedis(jedis);
    }

    @Override
    public T map(T obj) throws Exception {
        //根据流中对象获取要关联的维度主键
        String rowKey = getRowKey(obj);
        //根据维度的主键到Redis中获取维度数据
        JSONObject dimJsonObj = RedisUtil.readDim(jedis, getTableName(), rowKey);
        if(dimJsonObj != null){
            //如果在Redis中获取到了要关联的维度，直接将其作为返回值返回(缓存命中)
            System.out.println("~~从Redis中获取到了"+getTableName()+"表的"+rowKey+"维度数据~~~");
        }else {
            //如果在Redis中没有获取到要关联的维度，发送请求到HBase中获取
            dimJsonObj = HBaseUtil.getRow(hbaseConn, Constant.HBASE_NAMESPACE, getTableName(), rowKey, JSONObject.class);
            if(dimJsonObj != null){
                System.out.println("~~从HBase中获取到了"+getTableName()+"表的"+rowKey+"维度数据~~~");
                //将从HBase中获取到的维度数据放到Redis中缓存起来，方便下次查询使用
                RedisUtil.writeDim(jedis, getTableName(), rowKey,dimJsonObj);
            }
        }

        if(dimJsonObj != null){
            //将维度属性补充到流中对象上
            addDims(obj, dimJsonObj);
        }
        return obj;
    }

}
